package com.settlement.system.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.settlement.system.common.constant.KafkaConstants;
import com.settlement.system.common.constant.SystemConstants;
import com.settlement.system.common.util.DateUtils;
import com.settlement.system.common.util.NumberUtils;
import com.settlement.system.common.util.StringUtils;
import com.settlement.system.handler.disruptor.MessageVo;
import com.settlement.system.mapper.WmFundFlowMapper;
import com.settlement.system.mapper.WmTradeDetailMapper;
import com.settlement.system.model.entity.*;
import com.settlement.system.model.form.WmFundFlowForm;
import com.settlement.system.model.query.WmFundFlowQuery;
import com.settlement.system.model.vo.WmFundFlowImportVo;
import com.settlement.system.service.*;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 微盟资金流水 服务实现类
 * </p>
 *
 * @author huangrensen
 * @since 2024-06-18
 */
@Service
@RequiredArgsConstructor
public class WmFundFlowServiceImpl extends ServiceImpl<WmFundFlowMapper, WmFundFlow> implements WmFundFlowService {

    private final Logger logger = LoggerFactory.getLogger(WmFundFlowServiceImpl.class);
    private final SysStoreService sysStoreService;

    private final RedisTemplate redisTemplate;

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final SysImportErrorDetailService importErrorDetailService;
    private final WmNoRecordService wmNoRecordService;

    private final WmTradeDetailMapper wmTradeDetailMapper;

    private final KafkaSendErrorRecordService kafkaSendErrorRecordService;
    private final SnowflakeIdWorker snowflakeIdWorker = SnowflakeIdWorker.getInstance();

    @Override
    public Page<WmFundFlow> getPage(WmFundFlowQuery queryParams) {
        // 查询参数
        int pageNum = queryParams.getPageNum();
        int pageSize = queryParams.getPageSize();
        // 拼接月份列表
        List<String> monthList = new ArrayList<>();
        String endMonthStr = queryParams.getEndMonth().replaceAll("-", "");
        LocalDate endMonth = LocalDate.parse(endMonthStr + "01", DateTimeFormatter.ofPattern("yyyyMMdd"));
        LocalDate startTime = LocalDate.parse(queryParams.getStartMonth().replaceAll("-", "") + "01", DateTimeFormatter.ofPattern(
                "yyyyMMdd"));
        // 获取已经生成的表名
        Object object = redisTemplate.opsForValue().get(SystemConstants.WM_FUND_FLOW_TABLE);
        List<String> tableNameList;
        if (object != null) {
            tableNameList = (List<String>) object;
        } else {
            tableNameList = this.baseMapper.getAllTable();
            // 添加缓存
            redisTemplate.opsForValue().set(SystemConstants.WM_FUND_FLOW_TABLE, tableNameList, 1, TimeUnit.HOURS);
        }
        // 循环追加月份
        while (!startTime.isAfter(endMonth)) {
            monthList.add(startTime.format(DateTimeFormatter.ofPattern("yyyyMM")));
            startTime = startTime.plusMonths(1);
        }
        // 根据月份判断数据库表是否存在
        List<String> existTableList = new ArrayList<>();
        String tableName;
        for (String month : monthList) {
            tableName = "wm_fund_flow_%s".formatted(month);
            if (tableNameList.contains(tableName)) {
                existTableList.add(tableName);
            }
        }
        // 查询数据
        Page<WmFundFlow> page = new Page<>(pageNum, pageSize);
        if (existTableList.size() == 0) {
            return page;
        }
        queryParams.setTableNameList(existTableList);
        Page<WmFundFlow> mapperPage = this.baseMapper.getPage(page, queryParams);
        SysStore sysStore;
        for (WmFundFlow record : mapperPage.getRecords()) {
            sysStore = sysStoreService.getById(record.getStoreId());
            if (Objects.nonNull(sysStore)) {
                record.setStoreName(sysStore.getCommercePlatformStoreName());
            }
        }
        return mapperPage;
    }

    @Override
    public boolean saveWmFundFlow(WmFundFlowForm wmFundFlowForm) {
        WmFundFlow wmFundFlow = new WmFundFlow();
        BeanUtils.copyProperties(wmFundFlowForm, wmFundFlow);
        wmFundFlow.setCreateTime(null);
        return this.saveOrUpdate(wmFundFlow);
    }

    @Override
    public WmFundFlowForm getWmFundFlowFormData(Long id) {
        WmFundFlow wmFundFlow = this.getById(id);
        WmFundFlowForm wmFundFlowForm = new WmFundFlowForm();
        BeanUtils.copyProperties(wmFundFlow, wmFundFlowForm);
        return wmFundFlowForm;
    }

    @Override
    public boolean deleteWmFundFlow(String ids) {
        Assert.isTrue(StringUtils.hasText(ids), "删除的数据为空");
        List<Long> idArray = Arrays.stream(ids.split(",")).map(Long::parseLong).collect(Collectors.toList());
        return this.removeByIds(idArray);
    }

    @KafkaListener(id = KafkaConstants.HWG_TOPIC_WM_FUND_FLOW, topics = KafkaConstants.HWG_TOPIC_WM_FUND_FLOW, groupId = "yzConsumerGroup")
    public void importData(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        logger.info("【{}】当前消费者本次拉取的数据量为：{}", KafkaConstants.HWG_TOPIC_WM_FUND_FLOW, (Objects.isNull(records)? 0 : records.size()));
        if(Objects.isNull(records) || records.isEmpty()){
            logger.error("【{}】无消费数据...", KafkaConstants.HWG_TOPIC_WM_FUND_FLOW);
        }
        String orderNo = "";
        for (ConsumerRecord<String, String> record : records) {
            String importVoStr = record.value();
            try {
                WmFundFlowImportVo importVo = JSONObject.parseObject(importVoStr, WmFundFlowImportVo.class);
                orderNo = importVo.getFlowNo();
                this.importData(Collections.singletonList(importVo));
            } catch (Exception e) {
                logger.info("消费失败：{}", e.getMessage());
                // 保存失败数据
                SysImportErrorDetail detail = new SysImportErrorDetail();
                detail.setPlatformAndType("微盟资金流水");
                detail.setDataText(importVoStr);
                detail.setOrderNo(orderNo);
                detail.setQueueName(KafkaConstants.HWG_TOPIC_WM_FUND_FLOW);
                detail.setRemark(e.getMessage());
                importErrorDetailService.checkAndCreate(detail);
            }
        }
        // 确认消费
        acknowledgment.acknowledge();
    }

    @Override
    public void importData(List<WmFundFlowImportVo> voList) {
        redisTemplate.delete(SystemConstants.WM_FUND_FLOW_TABLE);
        List<String> billMonthList = voList.stream().map(WmFundFlowImportVo::getBillMonth).distinct().toList();
        // 创建表
        for (String billMonth : billMonthList) {
            if (!StringUtils.hasText(this.baseMapper.existTable("wm_fund_flow_%s".formatted(billMonth)))) {
                this.baseMapper.createTable("wm_fund_flow_%s".formatted(billMonth));
            }
        }
        // 获取微信支付业务单号
        List<String> wxBusinessNoList = voList.stream().map(WmFundFlowImportVo::getWxBusinessNo).filter(StringUtils::hasText).distinct().toList();
        // 查出关联记录
        List<WmNoRecord> wmNoRecordList = wmNoRecordService.list(new LambdaQueryWrapper<WmNoRecord>().in(WmNoRecord::getTradeNo, wxBusinessNoList));
        // 找出月份
        List<String> monthList = new ArrayList<>();
        for (WmNoRecord wmNoRecord : wmNoRecordList) {
            if (!monthList.contains(wmNoRecord.getTableId().substring(0, 6))) {
                monthList.add(wmNoRecord.getTableId().substring(0, 6));
            }
        }
        // 取出ID集合
        List<String> wmTradeDetailIdList = wmNoRecordList.stream().map(WmNoRecord::getTableId).toList();
        List<WmTradeDetail> tradeDetails = new ArrayList<>();
        WmTradeDetail query;
        for (String month : monthList) {
            query = new WmTradeDetail();
            query.setIdList(wmTradeDetailIdList);
            query.setTableName("wm_trade_detail_%s".formatted(month));
            List<WmTradeDetail> details = wmTradeDetailMapper.getList(query);
            if (details.size() > 0) {
                tradeDetails.addAll(details);
            }
        }
        // 创建集合
        List<WmFundFlow> wmFundFlowList = new ArrayList<>();
        for (WmFundFlowImportVo importVo : voList) {
            WmFundFlow wmFundFlow = new WmFundFlow();
            BeanUtils.copyProperties(importVo, wmFundFlow);
            wmFundFlow.setId(snowflakeIdWorker.nextId(importVo.getBillMonth()));
            // 记账时间
            wmFundFlow.setBillTime(DateUtils.convertStringToLocalDateTime(importVo.getBillTime()));
            // 收支金额(元)
            wmFundFlow.setIncomeAmount(NumberUtils.convertBigDecimal(importVo.getIncomeAmount()));
            // 账户结余(元)
            wmFundFlow.setBalanceAmount(NumberUtils.convertBigDecimal(importVo.getBalanceAmount()));
            wmFundFlow.setTableName("wm_fund_flow_%s".formatted(importVo.getBillMonth()));
            wmFundFlow.setStoreId(importVo.getStoreId());
            // 填充交易单号和销售门店
            query = tradeDetails.stream().filter(item -> StringUtils.hasText(wmFundFlow.getWxBusinessNo())
                    && wmFundFlow.getWxBusinessNo().contains(item.getOtherTradeNo())).findFirst().orElse(null);
            if (Objects.nonNull(query)) {
                wmFundFlow.setTradeNo(query.getTradeNo());
                wmFundFlow.setSaleStore(query.getSellStore());
            }
            wmFundFlowList.add(wmFundFlow);
        }
        MessageVo vo;
        List<WmFundFlow> itemList = new ArrayList<>();
        // 创建
        for (WmFundFlow wmFundFlow : wmFundFlowList) {
            wmFundFlow.cutString();
            this.baseMapper.insertData(wmFundFlow);
            itemList.add(wmFundFlow);
        }
        // 发送消息
        for (WmFundFlow wmFundFlow : itemList) {
            // 发送通知
            vo = new MessageVo();
            vo.setDataId(wmFundFlow.getId());
            vo.setStoreId(wmFundFlow.getStoreId());
            vo.setBillMonth(wmFundFlow.getBillMonth());
            // 根据字典中的备注查找对应的数据库表字段名
            vo.setTableNamePrefix("wm_fund_flow_");
            // 从字段管理里找到对应的字段类型编码，获取对应的billType
            vo.setDictCode("wmFundFlow");
            // 推送到转换标准账单的消息队列
            vo.setUuid(snowflakeIdWorker.nextId());
            CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_BILL_QUEUE, JSONObject.toJSONString(vo));
            MessageVo finalVo = vo;
            completableFuture.whenComplete((result, throwable) -> {
                if(Objects.nonNull(throwable)){
                    throwable.printStackTrace();
                    KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                    errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                    errorRecord.setUuid(finalVo.getUuid());
                    errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_BILL_QUEUE);
                    errorRecord.setRemark(throwable.getMessage());
                    kafkaSendErrorRecordService.checkAndSave(errorRecord);
                }
            });
        }
    }

    @Override
    public List<WmFundFlowImportVo> findByIds(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return new ArrayList<>();
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<WmFundFlow> wmFundFlowList = this.getBaseMapper().selectBatchIds(ids);
        List<WmFundFlowImportVo> voList = new ArrayList<>();
        for (WmFundFlow wmFundFlow : wmFundFlowList) {
            WmFundFlowImportVo vo = new WmFundFlowImportVo();
            BeanUtils.copyProperties(wmFundFlow, vo);
            voList.add(vo);
        }
        return voList;
    }
}
